package handlers

import (
	"encoding/json"
	"fmt"
	"generalhandler/conf"
	"github.com/go-redis/redis"
	"github.com/nsqio/go-nsq"
	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
	"strconv"
	"strings"
	"time"
)

type MQHandler struct {
	BaseMQHandler
	handler Handler
}

type redisData struct {
	Tkid      string `json:"tkid"`
	SubDomain string `json:"subdomain"`
	Port      string `json:"port"`
}

type RedisHandler struct {
	redisClient *redis.Client
}

var instance *RedisHandler

func RedisHandlerInstance() *RedisHandler {

	return instance
}

func init() {
	fmt.Println("init in mq ")
	instance = new(RedisHandler)
	instance.redisClient = redis.NewClient(&redis.Options{
		Addr:     "redis_1:6379",
		Password: "", //默认空密码
		DB:       0,  //使用默认数据库
	})
}

type Store struct {
	session *mgo.Session
}

var s *Store

func StoreInstance() *Store {
	if nil != s {
		return s
	}
	conf.LoadStoreConf()
	s = new(Store)
	err := s.Init()
	if nil != err {
		panic(err)
	}
	return s
}

func (s *Store) Init() error {
	var err error
	s.session, err = mgo.Dial(conf.StoreConfigInstance().StoreAddress)
	if nil != err {
		return fmt.Errorf("failed to mgo.Dial, cuz %s", err.Error())
	}
	s.session.SetMode(mgo.Monotonic, true)
	credential := &mgo.Credential{Username: conf.StoreConfigInstance().StoreUser,
		Password: conf.StoreConfigInstance().StorePwd}
	if err := s.session.Login(credential); err != nil {
		return fmt.Errorf("failed to mgo.Login, cuz %s", err.Error())
	}
	return nil
}

func (s *Store) Session() *mgo.Session {
	return s.session.Clone()
}

func MQHandlerInstance(handler Handler) *MQHandler {
	instance := new(MQHandler)
	instance.SetHandler(handler)
	instance.Init()
	return instance
}

func (mqh *MQHandler) SetHandler(handler Handler) {
	mqh.handler = handler
}

func (mqh *MQHandler) HandleMessage(msg *nsq.Message) error {
	msg.DisableAutoResponse()
	go func(msg *nsq.Message) {
		if err := mqh.handleMessage(msg.Body); nil != err {
			msg.Requeue(-1)
		} else {
			msg.Finish()
		}
	}(msg)

	return nil
}

func (mqh *MQHandler) PublishMessage(topic string, msg []byte) {
	var data redisData
	err := json.Unmarshal(msg, &data)
	if nil != err {
		mqh.Warn("failed to unmarshal %s, cuz %s", string(msg), err.Error())
		return
	}
	finishFlag, err := RedisHandlerInstance().redisClient.Get(data.Tkid).Result()
	if err != nil {
		int, err := strconv.Atoi(finishFlag)
		if err != nil && int >= 9990000 { //停止标志-10000
			return
		}
	}
	handRedis(data, mqh, topic, 0)
	mqh.BaseMQHandler.PublishMessage(topic, msg)
}

func (mqh *MQHandler) handleMessage(bytesMsg []byte) error {
	var data redisData
	err := json.Unmarshal(bytesMsg, &data)
	if nil != err {
		mqh.Warn("failed to unmarshal %s, cuz %s", string(bytesMsg), err.Error())
		return err
	}
	handRedis(data, mqh, conf.MQConfigInstance().MQTopic, 2)
	err = mqh.handler.Handle(bytesMsg, mqh, mqh)
	handRedis(data, mqh, conf.MQConfigInstance().MQTopic, 1)
	return err
}

func recordTime(topic string, lastKeyStr string, keyStr string, status int, log Logger, redisMap map[string]interface{}) {
	if topic != "store" {
		log.Info("last %s %s", lastKeyStr, keyStr)
		if status == 0 {
			redisMap["publishTime"] = time.Now().Unix()
		} else if status == 2 {
			_, err := RedisHandlerInstance().redisClient.Get(keyStr).Result()
			if nil == err {
				//说明已有开始记录，即记录了发布时间和开始时间，就不需要在用他更新时间记录了
				_, err = RedisHandlerInstance().redisClient.Del(lastKeyStr).Result()
				if nil != err {
					log.Info("Del(lastKeyStr) %s %s", lastKeyStr, err.Error())
				}
				return
			} else {
				redisMap["startTime"] = time.Now().Unix()
				lastScan, err := RedisHandlerInstance().redisClient.Get(lastKeyStr).Result()
				if nil == err {
					var lastJson map[string]interface{}
					if err = json.Unmarshal([]byte(lastScan), &lastJson); err == nil {
						if _, ok := lastJson["publishTime"]; ok {
							redisMap["publishTime"] = lastJson["publishTime"]
						}

					} else {
						log.Warn("json.Unmarshal publishTime error %s", err.Error())
					}
					_, err = RedisHandlerInstance().redisClient.Del(lastKeyStr).Result()
					if nil != err {
						log.Info("Del(lastKeyStr) %s %s", lastKeyStr, err.Error())
					}
				} else {
					redisMap["publishTime"] = time.Now().Unix() //起始节点没有publishTime
				}
			}
		} else if status == 1 {
			redisMap["finishTime"] = time.Now().Unix()
			currScan, err := RedisHandlerInstance().redisClient.Get(keyStr).Result()
			if nil == err {
				//说明已有结束记录，即记录了发布时间和开始时间，只需要更新结束时间记录
				var currJson map[string]interface{}
				if err = json.Unmarshal([]byte(currScan), &currJson); err == nil {
					redisMap["publishTime"] = currJson["publishTime"]
					redisMap["startTime"] = currJson["startTime"]
				} else {
					log.Warn("json.Unmarshal keyStr error %s", err.Error())
				}
			} else {
				lastScan, err := RedisHandlerInstance().redisClient.Get(lastKeyStr).Result()
				if nil != err {
					log.Warn("Gel(lastKeyStr) %s %s", lastKeyStr, err.Error())
				} else {
					var lastJson map[string]interface{}
					if err = json.Unmarshal([]byte(lastScan), &lastJson); err == nil {
						redisMap["publishTime"] = lastJson["publishTime"]
						redisMap["startTime"] = lastJson["startTime"]
					} else {
						log.Warn("json.Unmarshal lastKeyStr error %s", err.Error())
					}
				}
			}
			_, err = RedisHandlerInstance().redisClient.Del(lastKeyStr).Result()
			if nil != err {
				log.Info("Del(lastKeyStr) %s %s", lastKeyStr, err.Error())
			}
		}
		redisJson, err := json.Marshal(redisMap)
		if err != nil {
			log.Info("redisJson Marshal error %s", err.Error())
		} else {
			_, err = RedisHandlerInstance().redisClient.Set(keyStr, redisJson, time.Hour*24*7).Result()
			if err != nil {
				log.Info("redisClient.Set error %s", err.Error())
			}
		}
	}
}

func handRedis(data redisData, log Logger, topic string, status int) {
	tkidCount, err := RedisHandlerInstance().redisClient.Get(data.Tkid).Result()
	if nil == err {
		finishFlag, err := strconv.Atoi(tkidCount)
		if err == nil {
			if finishFlag > 9990000 { //停止标志-10000
				return
			}
		}
	}
	if 0 == len(data.Tkid) {
		log.Warn("found no tkid in msg %s", topic)
		return
	}

	lastStatus := 0
	if status == 1 {
		lastStatus = 2
	}
	var redisMap map[string]interface{}
	redisMap = make(map[string]interface{})
	keyStr := fmt.Sprintf("%s-%s:%d", data.Tkid, topic, status)
	redisMap["topic"] = topic
	lastKeyStr := fmt.Sprintf("%s-%s:%d", data.Tkid, topic, lastStatus)
	if 0 != len(data.SubDomain) {
		keyStr = fmt.Sprintf("%s-%s-%s:%d", data.Tkid, topic, data.SubDomain, status)
		redisMap["subdomain"] = data.SubDomain
		lastKeyStr = fmt.Sprintf("%s-%s-%s:%d", data.Tkid, topic, data.SubDomain, lastStatus)
	}
	if 0 != len(data.Port) {
		keyStr = fmt.Sprintf("%s-%s-%s-%s:%d", data.Tkid, topic, data.SubDomain, data.Port, status)
		redisMap["port"] = data.Port
		lastKeyStr = fmt.Sprintf("%s-%s-%s-%s:%d", data.Tkid, topic, data.SubDomain, data.Port, lastStatus)
	}
	recordTime(topic, lastKeyStr, keyStr, status, log, redisMap)

	if status == 0 {
		finishFlag, err := RedisHandlerInstance().redisClient.Incr(data.Tkid).Result()
		if err != nil {
			log.Info("Incri error %s", err.Error())
		} else {
			log.Info("pub %s--%d", topic, finishFlag)
		}
	} else if status == 1 {
		finishFlag, err := RedisHandlerInstance().redisClient.Decr(data.Tkid).Result()
		if err == nil {
			log.Info("handler %s--%d", topic, finishFlag)
			if finishFlag == 0 {
				table := StoreInstance().Session().DB(conf.StoreConfigInstance().StoreDB).C("scantask")
				if err := table.Update(bson.M{"tkid": data.Tkid}, bson.M{"$set": bson.M{"finished": 1}}); err != nil {
					log.Error("failed to update%s, cuz %s", data.Tkid, err.Error())
					return
				}
				var cursor uint64
				var val []string
				for {
					var keys []string
					var err error
					keys, cursor, err = RedisHandlerInstance().redisClient.Scan(cursor, data.Tkid+"*", 0).Result()
					if err != nil {
						log.Error("failed to redisClient.Scan , cuz %s,%s", data.Tkid, err.Error())
						continue
					}
					val = append(val, keys...)
					if cursor == 0 {
						break
					}
				}
				if err != nil {
					log.Error("failed to get data.Tkid , cuz %s,%s", data.Tkid, err.Error())
					return
				}
				if err := table.Update(bson.M{"tkid": data.Tkid}, bson.M{"$set": bson.M{"scanredis": val}}); err != nil {
					log.Error("failed to update, cuz %s,%s", data.Tkid, err.Error())
					return
				}
				var arr []interface{}
				for k, _ := range val {
					value, _ := RedisHandlerInstance().redisClient.Get(val[k]).Result()
					if sum := strings.Index(value, "{"); sum != -1 {
						var jsonValue map[string]interface{}
						if err = json.Unmarshal([]byte(value), &jsonValue); err == nil {
							arr = append(arr, jsonValue)
						} else {
							log.Warn("json.Unmarshal error %s", err.Error())
						}
					}
					RedisHandlerInstance().redisClient.Del(val[k])
				}
				if err := table.Update(bson.M{"tkid": data.Tkid}, bson.M{"$set": bson.M{"scanredis": arr}}); err != nil {
					log.Error("failed to update, cuz %s,%s", data.Tkid, err.Error())
					return
				}
			}
		} else {
			log.Warn("Decr error %s", err.Error())
		}
	} else {
		log.Info("start handler %s--%d", topic, 2)
	}
}
